package com.paradogs.framework.core.queue;

import com.paradogs.framework.core.msg.ByteData;
import com.paradogs.framework.core.msg.PRMsg;
import com.paradogs.framework.core.utils.RandomUtils;
import com.paradogs.framework.core.utils.SpringUtils;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: yumi
 * @date: 2023/7/27  14:22
 * @Description: 队列管理器
 */
@Slf4j
public class PRQueueHolder implements ApplicationRunner {

    @Getter
    @Setter
    @Autowired
    private QueueProperties queueProperties;

    /**
     * 队列集合
     * 待优化：有一些业务不需要保证顺序性，可以创建一个队列由多个消费者消费，不需要一对一，提高效率
     */
    private static Map<QueueType, List<QueueConsumerThread>> consumerMap = new ConcurrentHashMap<>();

    /**
     * 初始化相关队列
     * ps: 这里要用自定义方法不能用构造器，因为构造器 properties 还未注入完成（空指针）
     */
    public void init() {
        initQueue(QueueType.MAIN, queueProperties.getMainSize());
        initQueue(QueueType.DB, queueProperties.getDbSize());
    }

    public void initQueue(QueueType type, int size) {
        if (size == 0) {return;}
        for (int i = 0; i < size; i++) {
            // 原型模式 getBean
            QueueConsumerThread queueConsumer = SpringUtils.getBean(type.queueConsumerThreadClass);
            if (i == 0) { // 第一个线程额外负责处理循环检测
                queueConsumer.setLoop(true);
            }
            queueConsumer.setKey(type.key + "-" + (i + 1));
            try {
                queueConsumer.start();
            } catch (IllegalThreadStateException e) {
                throw new RuntimeException("Consumer[" + queueConsumer.getClass().getName() + "] is singleton, please use @Scope(\"prototype\")");
            }
            if (!consumerMap.containsKey(type)) {
                consumerMap.put(type, new ArrayList<>());
            }
            consumerMap.get(type).add(queueConsumer);
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.init();
    }

    public enum QueueType {

        /**
         * 消息逻辑处理
         */
        MAIN(AbstractMsgConsumer.class, "MainMsgConsumer"),
        /**
         * DB 持久化处理
         */
        DB(DBConsumer.class, "DBConsumer");

        /**
         * 消费对象类型
         */
        private Class<? extends QueueConsumerThread> queueConsumerThreadClass;
        /**
         * 线程描述标识符
         */
        private String key;

        QueueType(Class<? extends QueueConsumerThread> queueConsumerThreadClass, String key) {
            this.queueConsumerThreadClass = queueConsumerThreadClass;
            this.key = key;
        }
    }

    public static void putMsg(QueueType queueType, ByteData msg) throws InterruptedException {
        balance(consumerMap.get(queueType), msg).produce(msg);
    }
    public static void putMsg(QueueType queueType, DBUpdatMsg msg) {
        balance(consumerMap.get(queueType), msg).produce(msg);
    }

    /**
     * 待优化，负载均衡规则，可以换成一个接口
     * @return
     */
    private static QueueConsumerThread balance(List<QueueConsumerThread> list, ByteData msg) {
        if (msg.getMsgHeader().getPId() == null) {
            return list.get(0);
        }
        return list.get((int) (msg.getMsgHeader().getPId() % list.size()));
    }
    private static QueueConsumerThread balance(List<QueueConsumerThread> list, DBUpdatMsg msg) {
        if (msg.getPId() == null) {
            return list.get(0);
        }
        return list.get((int) (msg.getPId() % list.size()));
    }



}
